package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.bean.TrafficUniqueVisitorBean;
import com.atguigu.edu.realtime.common.kafkaTopics;
import com.atguigu.edu.realtime.util.ClickHouseUtil;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author Lec
 * @date 2022/9/3 19:45
 */

public class DwsTrafficUniqueVisitorWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //TODO 2. 设置状态后端
        /*
        env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
        env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        */
        String groupId = "dws_traffic_unique_visitor_window";
        //TODO 3. 读取kafka对应主题数据
        DataStreamSource<String> streamSource = env.addSource(KafkaUtil.getKafkaConsumer(kafkaTopics.DWD_TRAFFIC_UNIQUE_VISITOR_DETAIL, groupId));

        //TODO 4.转换结构为Javabean
        SingleOutputStreamOperator<TrafficUniqueVisitorBean> beanStream = streamSource.map(new MapFunction<String, TrafficUniqueVisitorBean>() {
            @Override
            public TrafficUniqueVisitorBean map(String value) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                String sc = common.getString("sc");
                String isNew = common.getString("is_new");
                Long ts = jsonObject.getLong("ts");
                return TrafficUniqueVisitorBean.builder()
                        .sc(sc)
                        .isNew(isNew)
                        .uvCt(1L)
                        .ts(ts)
                        .build();
            }
        });
        
        //TODO 5.添加水位线
        SingleOutputStreamOperator<TrafficUniqueVisitorBean> watermarkStream = beanStream.assignTimestampsAndWatermarks(WatermarkStrategy.<TrafficUniqueVisitorBean>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner(new SerializableTimestampAssigner<TrafficUniqueVisitorBean>() {
            @Override
            public long extractTimestamp(TrafficUniqueVisitorBean element, long recordTimestamp) {
                return element.getTs();
            }
        }));

        //TODO 6.开窗聚合
        KeyedStream<TrafficUniqueVisitorBean, String> keyedStream = watermarkStream.keyBy(new KeySelector<TrafficUniqueVisitorBean, String>() {
            @Override
            public String getKey(TrafficUniqueVisitorBean value) throws Exception {
                String key = value.getSc() + value.getIsNew();
                return key;
            }
        });

        WindowedStream<TrafficUniqueVisitorBean, String, TimeWindow> windowedStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10L)));

        SingleOutputStreamOperator<TrafficUniqueVisitorBean> resultStream = windowedStream.reduce(new ReduceFunction<TrafficUniqueVisitorBean>() {
            @Override
            public TrafficUniqueVisitorBean reduce(TrafficUniqueVisitorBean value1, TrafficUniqueVisitorBean value2) throws Exception {
                value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                return value1;
            }
        }, new WindowFunction<TrafficUniqueVisitorBean, TrafficUniqueVisitorBean, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<TrafficUniqueVisitorBean> input, Collector<TrafficUniqueVisitorBean> out) throws Exception {
                TrafficUniqueVisitorBean trafficUniqueVisitorBean = input.iterator().next();
                trafficUniqueVisitorBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                trafficUniqueVisitorBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                out.collect(trafficUniqueVisitorBean);
            }
        });
//        resultStream.print();

        //TODO 7.将结果写入到clickhouse中
        resultStream.addSink(ClickHouseUtil.getClickHouseSinkFunc("insert into table dws_traffic_unique_visitor_window values(?,?,?,?,?,?)"));

        //TODO 8. 执行任务
        env.execute(groupId);

    }
}
